package org.example.udf;

import org.apache.hadoop.hive.ql.exec.UDF;
import org.apache.hadoop.io.Text;
import java.util.*;

/**
 * 员工组织架构递归UDF：获取指定员工的所有上级
 * 输入：员工ID，员工-上级映射关系(格式: "员工ID:上级ID,员工ID:上级ID")
 * 输出：所有上级ID，用逗号分隔(从直接上级到顶级)
 */
public class EmployeeHierarchyUDF extends UDF {

    public static void main(String[] args) {
        // 创建UDF实例
        EmployeeHierarchyUDF udf = new EmployeeHierarchyUDF();

        // 样本数据1：正常多层级架构
        System.out.println("=== 测试案例1：正常多层级 ===");
        String empId1 = "1001";  // 员工ID
        // 员工-上级映射：格式 "员工ID:上级ID,员工ID:上级ID"
        String mapStr1 = "1001:1003,1002:1003,1003:1005,1004:1005,1005:1007,1006:1007,1007:1007";
        Text result1 = udf.evaluate(new Text(empId1), new Text(mapStr1));
        System.out.println("员工1001的所有上级：" + result1.toString());  // 预期：1003,1005,1007
    }

    // 主方法：处理输入并返回结果
    public Text evaluate(Text employeeId, Text employeeMapStr) {
        // 处理空值
        if (employeeId == null || employeeMapStr == null) {
            return null;
        }

        // 解析员工-上级映射关系为Map
        Map<String, String> employeeMap = parseEmployeeMap(employeeMapStr.toString());

        // 递归获取所有上级
        List<String> allManagers = new ArrayList<>();
        findAllManagers(employeeId.toString(), employeeMap, allManagers);

        // 转换为逗号分隔的字符串返回
        return new Text(String.join(",", allManagers));
    }

    // 解析员工-上级映射字符串为Map
    private Map<String, String> parseEmployeeMap(String mapStr) {
        Map<String, String> map = new HashMap<>();
        if (mapStr == null || mapStr.trim().isEmpty()) {
            return map;
        }

        String[] entries = mapStr.split(",");
        for (String entry : entries) {
            String[] parts = entry.split(":");
            if (parts.length == 2) {
                map.put(parts[0].trim(), parts[1].trim());
            }
        }
        return map;
    }

    // 递归查找所有上级
    private void findAllManagers(String employeeId, Map<String, String> employeeMap, List<String> result) {
        // 获取直接上级
        String managerId = employeeMap.get(employeeId);

        // 终止条件：没有上级或已经找到顶级
        if (managerId == null || managerId.isEmpty() || managerId.equals(employeeId)) {
            return;
        }

        // 检查是否存在循环引用（防止无限递归）
        if (result.contains(managerId)) {
            throw new IllegalArgumentException("组织架构中存在循环引用: " + managerId);
        }

        // 添加当前上级到结果
        result.add(managerId);

        // 递归查找上级的上级
        findAllManagers(managerId, employeeMap, result);
    }
}
